package com.wk.springbootlearn.kafka;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Component
public class KafkaProducer implements CommandLineRunner {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void run(String... args) throws Exception {
//        // kafka生产者 测试使用
//        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
//            kafkaTemplate.send("hello", String.valueOf(System.currentTimeMillis()))
//                    .addCallback(new SuccessCallback<SendResult<String, String>>() {
//                        @Override
//                        public void onSuccess(SendResult<String, String> stringStringSendResult) {
//                            if (stringStringSendResult.getRecordMetadata() != null) {
//                                System.out.println("消息发送成功 offset: " + stringStringSendResult.getRecordMetadata().offset());
//                                return;
//                            }
//                            System.out.println("消息发送成功");
//                        }
//                    }, new FailureCallback() {
//                        @Override
//                        public void onFailure(Throwable throwable) {
//                            System.out.println("消息发送失败：" + throwable.getMessage());
//                        }
//                    });
//        },0,1, TimeUnit.SECONDS);
    }
}
